﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace RabbitMQExtension
{
    /// <summary>
    /// MQConsumerClient
    /// RabbitMQ消费客户端。
    /// 
    /// 修改纪录
    /// 
    ///		2020.04.23 版本：1.0 Sunnyfish 创建文件。
    /// 
    /// 版本：1.0
    /// 
    /// <author>
    ///		<name>Sunnyfish</name>
    ///		<date>2020.04.23</date>
    /// </author> 
    /// </summary>
    public class MQConsumerClient : IMQConsumerClient
    {
        /// <summary>
        /// 日记打印Log，可以外部传入接管默认Log
        /// </summary>
        public ILog Log { get; set; } = LogManager.GetLogger(typeof(MQConsumerClient));

        private object _ObjLock = new object();
        private string _Topices = "";
        private string _FilterKeys = "";
        private string _QueueName = "";
        /// <summary>
        /// 消息交换类型。
        /// </summary>
        private string _ExchangeType = "";


        private MQConfig _MQConfig = null;
        private ConnectionFactory _Factory = null;
        private IConnection _Connection = null;

        /// <summary>
        /// 消息缓冲量，默认100。
        /// </summary>
        private ushort _BasciQos = 100;

        private MessageProcessCallback _MessageProcessCallback = null;

        /// <summary>
        /// 消息订阅通道，每个消息主题可以有多个通道（受最大线程数参数控制）。
        /// </summary>
        private Dictionary<string, List<IModel>> _Channels = new Dictionary<string, List<IModel>>();
        /// <summary>
        /// 每个消息主题的二级过滤内容。
        /// </summary>
        private Dictionary<string, string[]> _TopiceFilters = new Dictionary<string, string[]>();

        /// <summary>
        /// 任务取消控制器。
        /// </summary>
        private CancellationTokenSource _TaskTokenSource = new CancellationTokenSource();

        /// <summary>
        /// 设置消息队列的连接信息。
        /// </summary>
        /// <param name="config">信息实体</param>
        public void SetConnectInfo(MQConfig config)
        {
            _MQConfig = config;
        }

        /// <summary>
        /// 与服务器是否处于连接状态。
        /// </summary>
        /// <returns>true:已连接，false:未连接。</returns>
        public bool IsConnected()
        {
            bool isConnect = false;
            if (_Connection != null)
                isConnect = _Connection.IsOpen;
            return isConnect;
        }

        /// <summary>
        /// 开始接收指定消息队列的消息。。
        /// </summary>
        /// <param name="queueName">要消费的消息队列的名称。</param>
        /// <param name="mpc">消息处理回调函数。</param>
        public void StartConsumer(string queue, MessageProcessCallback mpc)
        {
            _QueueName = queue;
            _MessageProcessCallback = mpc;
            if (mpc == null)
                throw new ArgumentNullException(nameof(MessageProcessCallback), "必须指定 回调函数 mpc");
            try
            {
                if (_Factory == null)
                {
                    lock (_ObjLock)
                    {
                        if (_Factory == null)
                        {
                            _Channels.Clear();
                            ConnectToServer();
                            IModel mainChannel = null;
                            if (!_Channels.ContainsKey(_QueueName))
                            {
                                mainChannel = _Connection.CreateModel();
                                //每个消费者同一时间最多分发100条消息。
                                mainChannel.BasicQos(0, _BasciQos, false);
                                _Channels.Add(_QueueName, new List<IModel>());
                            }
                            //一个线程对应一个消息通道。
                            _Channels[_QueueName].Add(mainChannel);
                            for (var i = 0; i < _MQConfig.MaxComsumerThreads; i++)
                            {
                                IModel childChannel = _Connection.CreateModel();
                                childChannel.BasicQos(0, _BasciQos, false);
                                _Channels[_QueueName].Add(childChannel);
                            }
                            foreach (var channel in _Channels[_QueueName])
                            {
                                var comsumer = new EventingBasicConsumer(channel);

                                comsumer.Received += (model, ea) =>
                                {
                                    MessageProcess(channel, ea.Body, ea.ConsumerTag, ea.RoutingKey, ea.BasicProperties.MessageId, ea.DeliveryTag, _MessageProcessCallback);
                                };
                                channel.BasicConsume(queue: queue,
                                    autoAck: false,
                                    consumer: comsumer
                                    );
                            }
                        }
                    }
                }
            }
            catch (Exception e)
            {
                Log.Error(e);
                if (_Factory != null)
                    _Factory = null;

                if (_Connection != null)
                {
                    _Connection.Close();
                    _Connection.Dispose();
                    _Connection = null;
                }

                if (_Channels.ContainsKey(_QueueName))
                {
                    foreach (IModel channel in _Channels[_QueueName])
                    {
                        if (channel != null)
                        {
                            channel.Close();
                            channel.Dispose();
                        }
                    }
                    _Channels[_QueueName].Clear();
                }
            }

        }

        /// <summary>
        /// 开始接收Topice方式推送的消息。
        /// </summary>
        /// <param name="topices">要消费的消息队列标志，可以多个,多个之间以"||"分隔</param>
        /// <param name="filterKey">对消费的消息队列进行二次过滤，与<paramref name="topices"/>一一对应，以"||"分隔，如果要单个队列使用多个过滤条件，过滤条件之间使用","分隔。
        /// 都不过滤使用"*"或传空字符。</param>
        /// <param name="mpc">消息处理回调函数，把你处理消息的函数传进来。</param>
        public void StartTopicConsumer(string topices, string filterKey, MessageProcessCallback mpc)
        {
            StartConsumer(topices, filterKey, "topic", mpc);
        }

        /// <summary>
        /// 开始接收发布订阅方式推送的消息。
        /// </summary>
        /// <param name="topices">要消费的消息队列标志，可以多个,多个之间以"||"分隔</param>
        /// <param name="mpc">消息处理回调函数，把你处理消息的函数传进来。</param>
        public void StartPubSubConsumer(string topices, MessageProcessCallback mpc)
        {
            StartConsumer(topices, "", "fanout", mpc);
        }

        /// <summary>
        /// 开始消费（订阅）指定的消息。
        /// </summary>
        /// <param name="topices">要消费的消息队列标志，可以多个,多个之间以"||"分隔</param>
        /// <param name="filterKeys">对消费的消息队列进行二次过滤，与<paramref name="topices"/>一一对应，以"||"分隔，如果要单个队列使用多个过滤条件，过滤条件之间使用","分隔。
        /// 都不过滤使用"*"或传空字符。</param>
        /// <param name="mpc">消息处理回调函数，把你处理消息的函数传进来。</param>
        /// <param name="exchangeType">消息交换类型。"topic"：主题模式，支持多级过滤；"fanout"，发布/订阅模式，不支持再次过滤。</param>
        /// <param name="expTime">消息有效时间，毫秒。默认为三天，过期自动删除。</param>
        private void StartConsumer(string topices, string filterKeys, string exchangeType, MessageProcessCallback mpc)
        {
            _Topices = topices;
            _TopiceFilters.Clear();
            _FilterKeys = filterKeys;
            _MessageProcessCallback = mpc;
            _ExchangeType = exchangeType;

            if (mpc == null)
                throw new ArgumentNullException(nameof(MessageProcessCallback), "必须指定 回调函数 mpc");

            string[] theTopices = _Topices.Split("||".ToCharArray(), StringSplitOptions.RemoveEmptyEntries);
            if (theTopices.Length < 1)
                throw new Exception("必须指定 topices 参数。");

            if (string.IsNullOrEmpty(_FilterKeys) && exchangeType.Equals("topic"))
            {
                //如果没有指定子过滤内容并且消息接收方式为"topic"，则默认接收所有消息。
                _FilterKeys = "*";
            }

            if (_FilterKeys != "*")
            {
                if (!string.IsNullOrEmpty(_FilterKeys))
                {
                    //如果有指定二次过滤，必须与topices保持一一对应关系。
                    string[] express = _FilterKeys.Split("||".ToCharArray(), StringSplitOptions.RemoveEmptyEntries);
                    if (express.Length != _Topices.Length)
                        throw new Exception("subExpress 参数的||数量必须与 topices 参数一致");
                    foreach (string top in theTopices)
                    {
                        _TopiceFilters.Add(top, express);
                    }
                }
            }
            else
            {
                //订阅所有Topice
                foreach (string top in theTopices)
                {
                    _TopiceFilters.Add(top, new string[] { "#" });
                }
            }
            try
            {
                if (_Factory == null)
                {
                    lock (_ObjLock)
                    {
                        if (_Factory == null)
                        {
                            _Channels.Clear();
                            ConnectToServer();
                            //订阅指定的Topice
                            foreach (string top in theTopices)
                            {
                                IModel mainChannel = null;
                                if (!_Channels.ContainsKey(top))
                                {
                                    mainChannel = _Connection.CreateModel();
                                    //mainChannel.ExchangeDeclare(exchange: top, type: exchangeType, durable: true);
                                    //每个消费者同一时间最多分发10条消息。
                                    mainChannel.BasicQos(0, _BasciQos, false);
                                    _Channels.Add(top, new List<IModel>());
                                }

                                //一个线程对应一个消息通道。
                                _Channels[top].Add(mainChannel);
                                for (int ti = 1; ti < _MQConfig.MaxComsumerThreads; ti++)
                                {
                                    IModel childChanel = _Connection.CreateModel();
                                    childChanel.BasicQos(0, _BasciQos, false);
                                    _Channels[top].Add(childChanel);
                                }
                            }
                        }
                    }
                }
                StartConsumerTasks();
            }
            catch (Exception e)
            {
                Log.Error(e);
                if (_Factory != null)
                    _Factory = null;

                if (_Connection != null)
                {
                    _Connection.Close();
                    _Connection.Dispose();
                    _Connection = null;
                }

                foreach (string top in theTopices)
                {
                    if (_Channels.ContainsKey(top))
                    {
                        foreach (IModel channel in _Channels[top])
                        {
                            if (channel != null)
                            {
                                channel.Close();
                                channel.Dispose();
                            }
                        }
                    }
                }
                _Channels.Clear();
            }
        }

        /// <summary>
        /// 启动所有的消费任务。
        /// </summary>
        private void StartConsumerTasks()
        {
            foreach (KeyValuePair<string, List<IModel>> kvpChannel in _Channels)
            {
                List<IModel> channels = kvpChannel.Value;
                foreach (IModel channel in channels)
                {
                    string topic = kvpChannel.Key;
                    string queueName = string.Format("que_{0}_{1}", _MQConfig.UserName, topic);

                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        MessageProcess(channel, ea.Body, ea.ConsumerTag, ea.RoutingKey, ea.BasicProperties.MessageId,
                            ea.DeliveryTag, _MessageProcessCallback);
                    };

                    channel.BasicConsume(queue: queueName,
                                     autoAck: false,
                                     consumer: consumer);
                }
            }
        }

        /// <summary>
        /// 连接到服务器。
        /// </summary>
        private void ConnectToServer()
        {
            _Factory = new ConnectionFactory()
            {
                VirtualHost = string.IsNullOrEmpty(_MQConfig.VirtualHost) ? "/" : _MQConfig.VirtualHost,
                UserName = _MQConfig.UserName,
                Password = _MQConfig.Password,
                NetworkRecoveryInterval = new TimeSpan(0, 0, 5),   //每5秒钟检查一次网络连接。
                AutomaticRecoveryEnabled = true,                    //网络连接异常时自动恢复。
                RequestedHeartbeat = 30                             //心跳检测超时时间，30秒。
            };
            var amqps = new List<AmqpTcpEndpoint>();
            var items = _MQConfig.Host.Split(new char[] { ',', ';' }, StringSplitOptions.RemoveEmptyEntries);
            foreach (var item in items)
            {
                var hosts = item.Split(':');
                var port = 5672;
                if (hosts.Length > 1)
                    int.TryParse(hosts[1], out port);
                if (port < 1)
                    port = 5672;
                amqps.Add(new AmqpTcpEndpoint
                {
                    HostName = hosts[0],
                    Port = port,
                });
            }
            _Connection = _Factory.CreateConnection(amqps);
        }

        private void MessageProcess(IModel channel, byte[] msgBody, string consumerTag, string routingKey,
           string msgId, ulong deliveryTag, MessageProcessCallback mpc)
        {
            MessageContent content = new MessageContent();
            Acks ack = Acks.CommitMessage;
            try
            {
                var body = msgBody;
                string message = Encoding.UTF8.GetString(body);
                content.MessageBody = message;
                content.TopicFlag = consumerTag;
                content.Topic = routingKey;
                content.MessageID = msgId;
                content.MessageKey = msgId;
                ack = mpc(content);
            }
            catch (Exception e)
            {
                ack = Acks.RejectLater;
                Log.Error(e);
            }
            finally
            {
                switch (ack)
                {
                    case Acks.CommitMessage:
                        channel.BasicAck(deliveryTag, false);
                        break;
                    case Acks.ReconsumeLater:
                        channel.BasicNack(deliveryTag, false, true);
                        break;
                    case Acks.RejectLater:
                        channel.BasicReject(deliveryTag, true);
                        break;
                }
            }
        }

        /// <summary>
        /// 关闭所有连接，停止接收消息。
        /// </summary>
        public void Shutdown()
        {
            if (!_TaskTokenSource.IsCancellationRequested)
                _TaskTokenSource.Cancel(false);

            if (_Factory != null)
                _Factory.AutomaticRecoveryEnabled = false;
            if (_Channels != null)
            {
                foreach (KeyValuePair<string, List<IModel>> kvp in _Channels)
                {
                    List<IModel> chanels = kvp.Value;
                    foreach (IModel chanel in chanels)
                    {
                        try
                        {
                            if (chanel != null)
                            {
                                if (chanel.IsOpen)
                                    chanel.Close();
                                chanel.Dispose();
                            }
                        }
                        catch (Exception e)
                        {
                            Log.Error($"释放消息通道发生异常:{e}");
                        }
                    }
                }
                _Channels.Clear();
            }

            if (_Connection != null)
            {
                try
                {
                    if (_Connection.IsOpen)
                    {
                        _Connection.Close();
                    }
                    _Connection.Dispose();
                }
                catch (Exception e)
                {
                    Log.Error($"释放消息连接发生异常:{e}");
                }
            }
            _Factory = null;
        }

        /// <summary>
        /// 释放资源。
        /// </summary>
        public void Dispose()
        {
            Shutdown();
            _Factory = null;
            _TopiceFilters.Clear();
        }

        /// <summary>
        /// 拉取消息通道
        /// </summary>
        private Dictionary<string, IModel> _PullChannels = new Dictionary<string, IModel>();

        /// <summary>
        /// 从队列中拉取一条消息
        /// </summary>
        /// <param name="queue">队列名</param>
        /// <param name="mpc">回调函数</param>
        public void Pull(string queue, MessageProcessCallback mpc)
        {
            if (string.IsNullOrEmpty(queue))
            {
                throw new ArgumentNullException(nameof(queue));
            }
            if (mpc == null)
            {
                throw new ArgumentNullException(nameof(MessageProcessCallback), "回调函数 mpc 尚未声明");
            }
            if (!IsConnected())
            {
                ConnectToServer();
            }
            IModel channel = null;
            if (!_PullChannels.ContainsKey(queue))
            {
                lock (_ObjLock)
                {
                    if (!_PullChannels.ContainsKey(queue))
                    {
                        channel = _Connection.CreateModel();
                        channel.QueueDeclare(queue, true, false, false);
                        _PullChannels[queue] = channel;
                    }
                }
            }
            channel = _PullChannels[queue];
            ulong deliveryTag = 0;
            Acks acks = Acks.CommitMessage;
            try
            {
                BasicGetResult result = null;
                lock (_ObjLock)
                    result = channel.BasicGet(queue, false);
                var content = new MessageContent();
                if (result != null)
                {
                    content.MessageID = result.BasicProperties.MessageId;
                    content.MessageBody = Encoding.UTF8.GetString(result.Body);
                    content.Topic = result.RoutingKey;
                    content.MessageKey = content.MessageID;
                    deliveryTag = result.DeliveryTag;
                }
                acks = mpc(content);
            }
            catch (Exception e)
            {
                Log.Error(e);
            }
            finally
            {
                lock (_ObjLock)
                    switch (acks)
                    {
                        case Acks.CommitMessage:
                            channel.BasicAck(deliveryTag, false);
                            break;
                        case Acks.ReconsumeLater:
                            channel.BasicNack(deliveryTag, false, true);
                            break;
                        case Acks.RejectLater:
                            channel.BasicReject(deliveryTag, true);
                            break;
                    }
            }
        }

        /// <summary>
        /// 根据枚举获取主题类型
        /// </summary>
        /// <param name="type">交换机类型枚举</param>
        /// <returns></returns>
        private string GetExchangeTypeString(MQExchangeType type)
        {
            string typeStr;
            switch (type)
            {
                case MQExchangeType.Direct:
                    typeStr = ExchangeType.Direct;
                    break;
                case MQExchangeType.Topic:
                    typeStr = ExchangeType.Topic;
                    break;
                case MQExchangeType.Headers:
                    typeStr = ExchangeType.Headers;
                    break;
                default:
                    typeStr = ExchangeType.Fanout;
                    break;
            }
            return typeStr;
        }
    }
}
